package com.gzyj.flink.source.mysql;

import com.gzyj.flink.pojo.gps.MonDaoludengjiSet;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class MonDaoludengjiSetSource extends RichSourceFunction<MonDaoludengjiSet> {

    private PreparedStatement ps;
    private Connection connection;
    private boolean stop = false;


    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        try {
            Class.forName("com.mysql.cj.jdbc.Driver");
            ExecutionConfig.GlobalJobParameters globalJobParam = getRuntimeContext().getExecutionConfig().getGlobalJobParameters();

            Configuration globconf = (Configuration) globalJobParam;

            this.connection = DriverManager.getConnection(globconf.getString("mysql-url", null),
                    globconf.getString("mysql-user", "root"),
                    globconf.getString("mysql-password", "123456"));
            this.ps = this.connection.prepareStatement("select * from mon_daoludengji_set where enable=1");
        } catch (Exception e) {
            System.out.println("-----------mysql get connection has exception , msg = " + e.getMessage());
        }


    }

    @Override
    public void run(SourceContext<MonDaoludengjiSet> sourceContext) throws Exception {


        while (!this.stop) {

            ResultSet rs = this.ps.executeQuery();
            while (rs.next()) {


                MonDaoludengjiSet set = new MonDaoludengjiSet();
                set.setMid(rs.getInt("mid"));
                set.setVehid(rs.getString("vehid"));
                set.setGaosu(rs.getInt("gaosu"));
                set.setKuaisu(rs.getInt("kuaisu"));
                set.setGuodao(rs.getInt("guodao"));
                set.setShengdao(rs.getInt("shengdao"));
                set.setXiandao(rs.getInt("xiandao"));
                set.setXiangdao(rs.getInt("xiangdao"));
                set.setQita(rs.getInt("qita"));
                set.setEnable(rs.getInt("enable"));
                set.setDelFlag(rs.getString("del_flag"));
                set.setCreateBy(rs.getString("create_by"));
                set.setCreateTime(rs.getDate("create_time"));
                set.setUpdateBy(rs.getString("update_by"));
                set.setUpdateTime(rs.getDate("update_time"));


                sourceContext.collect(set);


            }


            Thread.sleep(30 * 60 * 1000);
        }


    }

    @Override
    public void cancel() {
        this.stop = true;
        try {

            if (connection != null) { //关闭连接和释放资源
                connection.close();
            }
            if (ps != null) {
                ps.close();
            }
        } catch (Exception e) {

        }
    }
}
